package com.zch.rpc.register.register;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

public class ZKServerTask implements Runnable {
    public Socket getClient() {
        return client;
    }

    public void setClient(Socket client) {
        this.client = client;
    }

    Socket client = null;
    public long latestHeartTime;//最近一次通信时间

    public long getLatestHeartTime() {
        return latestHeartTime;
    }

    public void setLatestHeartTime(long latestHeartTime) {
        this.latestHeartTime = latestHeartTime;
    }

    public ZKServerTask(Socket client) {
        this.client = client;
        latestHeartTime=System.currentTimeMillis();
    }

    public void run() {

                ObjectInputStream input = null;
                ObjectOutputStream output = null;
                try {
                    input = new ObjectInputStream(client.getInputStream());
                    //TODO　增加强转类型判断
                    Message msg = (Message) input.readObject();
                    //处理注册/订阅任务
                    if ("register".equals(msg.getType())) {//服务端注册
                        output = new ObjectOutputStream(client.getOutputStream());
                        register(msg, output);
                    } else if ("subscribe".equals(msg.getType())) {//客户端订阅
                        output = new ObjectOutputStream(client.getOutputStream());
                        subscribe(msg, output);
                    }else if("heart".equals(msg.getType())){//发送心跳
                        output = new ObjectOutputStream(client.getOutputStream());
                        keepAlive(msg, output);
                    }
                    latestHeartTime=System.currentTimeMillis();
                    Runnable runnable= new Runnable() {
                        @Override
                        public void run() {
                            while(true){
                                try {
                                    Thread.sleep(2000);
                                    if(client!=null&&!client.isClosed()){
                                        if((System.currentTimeMillis()-latestHeartTime)/1000>=5){
                                            System.out.println("超过五秒没有通信，服务端关闭通道");
                                            client.close();
                                        }
                                    }

                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    };
                    Thread thread=new Thread(runnable);
                    thread.start();

                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (output != null) {
                        try {
                            output.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    if (input != null) {
                        try {
                            input.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }

                }






    }


    private void register(Message msg, ObjectOutputStream output) throws IOException {
        String interfaceName = msg.getInterfaceName();
        // 保存RPC服务端的地址信息（"serviceimpl,localhost:2881"）
        List<String> addressList = ZkServer.serviceRegistry.get(interfaceName);
        if (! (addressList == null || addressList.isEmpty())) {
            addressList.add(msg.getServiceAddress());
        } else {
            addressList = new ArrayList<String>();
            addressList.add(msg.getServiceAddress());
            ZkServer.serviceRegistry.put(interfaceName, addressList);
        }

        System.out.println("服务："+interfaceName + " 【 " + msg.getServiceAddress()+"】 " + "注册成功");
        String template = "register success: %s 【 %s 】";
        String info = String.format(template, interfaceName, msg.getServiceAddress());
        output.writeUTF(info);
    }

    private void subscribe(Message msg, ObjectOutputStream output) throws IOException {
        String interfaceName = msg.getInterfaceName();
        List<String> addressList = ZkServer.serviceRegistry.get(interfaceName);
        msg.setAvailableServiceAddresses(addressList);
        output.writeObject(msg);
        System.out.println(interfaceName + "订阅成功");
    }
    //心跳回复
    private void keepAlive(Message msg,ObjectOutputStream output) throws IOException{
         String remoteAddress = msg.getServiceAddress();
        String info = "zookeeper服务端 心跳返回";
        output.writeUTF(info);
        System.out.println("收到客户端 "+remoteAddress+"发来的心跳，并回复成功");
    }
}
